/*
 * Copyright 2014 Semidata. All rights reserved.
 */
package com.semidata.rtd.engine.main;

import java.util.concurrent.atomic.AtomicLong;

import com.semidata.rtd.core.service.Service;

/**
 * @author shewei@semidata.com, She Wei
 * 
 */
public class Monitor implements Service {

	private Thread thread = null;
	private boolean quit = false;
	public AtomicLong totalJobCount = new AtomicLong();
	public AtomicLong errorJobCount = new AtomicLong();

	public AtomicLong totalJobTime = new AtomicLong();
	public AtomicLong errorJobTime = new AtomicLong();

	public AtomicLong totalTaskCount = new AtomicLong();
	public AtomicLong totalTask1Count = new AtomicLong();
	public AtomicLong totalTask2Count = new AtomicLong();
	public AtomicLong totalTask3Count = new AtomicLong();

	public AtomicLong totalTaskTime = new AtomicLong();
	public AtomicLong totalTask1Time = new AtomicLong();
	public AtomicLong totalTask2Time = new AtomicLong();
	public AtomicLong totalTask3Time = new AtomicLong();

	public AtomicLong errorTaskCount = new AtomicLong();
	public AtomicLong errorTask1Count = new AtomicLong();
	public AtomicLong errorTask2Count = new AtomicLong();
	public AtomicLong errorTask3Count = new AtomicLong();

	public AtomicLong errorTaskTime = new AtomicLong();
	public AtomicLong errorTask1Time = new AtomicLong();
	public AtomicLong errorTask2Time = new AtomicLong();
	public AtomicLong errorTask3Time = new AtomicLong();
	
	private long startTime = 0;

	private void print() {
		StringBuffer sb = new StringBuffer();

		long time = (System.currentTimeMillis() - startTime);

		double jobPerSecond = (double) (totalJobCount.get() * 1000)
		        / (double) (time == 0 ? 1 : time);
		
		double jobAverageTime = (double) totalJobTime.get()
		        / (double) (totalJobCount.get() == 0 ? 1 : totalJobCount.get());

		double jobErrorRate = (double) errorJobCount.get()
		        / (double) (totalJobCount.get() == 0 ? 1 : totalJobCount.get());
		
		double taskPerSecond = (double) (totalTaskCount.get() * 1000)
		        / (double) (time == 0 ? 1 : time);
		
		double taskAverageTime = (double) totalTaskTime.get()
		        / (double) (totalTaskCount.get() == 0 ? 1 : totalTaskCount.get());

		double taskErrorRate = (double) errorTaskCount.get()
		        / (double) (totalTaskCount.get() == 0 ? 1 : totalTaskCount.get());
		
		double task1PerSecond = (double) (totalTask1Count.get() * 1000)
		        / (double) (time == 0 ? 1 : time);
		
		double task1AverageTime = (double) totalTask1Time.get()
		        / (double) (totalTask1Count.get() == 0 ? 1 : totalTask1Count.get());

		double task1ErrorRate = (double) errorTask1Count.get()
		        / (double) (totalTask1Count.get() == 0 ? 1 : totalTask1Count.get());
		
		double task2PerSecond = (double) (totalTask2Count.get() * 1000)
		        / (double) (time == 0 ? 1 : time);
		
		double task2AverageTime = (double) totalTask2Time.get()
		        / (double) (totalTask2Count.get() == 0 ? 1 : totalTask2Count.get());

		double task2ErrorRate = (double) errorTask2Count.get()
		        / (double) (totalTask2Count.get() == 0 ? 1 : totalTask2Count.get());

		double task3PerSecond = (double) (totalTask3Count.get() * 1000)
		        / (double) (time == 0 ? 1 : time);
		
		double task3AverageTime = (double) totalTask3Time.get()
		        / (double) (totalTask3Count.get() == 0 ? 1 : totalTask3Count.get());

		double task3ErrorRate = (double) errorTask3Count.get()
		        / (double) (totalTask3Count.get() == 0 ? 1 : totalTask3Count.get());
		
		sb.append(time / 1000);
	
		sb.append(' ').append(Math.round(jobPerSecond * 100) / 100.0);
		sb.append(' ').append(Math.round(jobAverageTime * 100) / 100.0);
		sb.append(' ').append(Math.round(jobErrorRate * 100) / 100.0);
		
		sb.append(' ').append(Math.round(taskPerSecond * 100) / 100.0);
		sb.append(' ').append(Math.round(taskAverageTime * 100) / 100.0);
		sb.append(' ').append(Math.round(taskErrorRate * 100) / 100.0);
		
		sb.append(' ').append(Math.round(task1PerSecond * 100) / 100.0);
		sb.append(' ').append(Math.round(task1AverageTime * 100) / 100.0);
		sb.append(' ').append(Math.round(task1ErrorRate * 100) / 100.0);
		
		sb.append(' ').append(Math.round(task2PerSecond * 100) / 100.0);
		sb.append(' ').append(Math.round(task2AverageTime * 100) / 100.0);
		sb.append(' ').append(Math.round(task2ErrorRate * 100) / 100.0);
		
		sb.append(' ').append(Math.round(task3PerSecond * 100) / 100.0);
		sb.append(' ').append(Math.round(task3AverageTime * 100) / 100.0);
		sb.append(' ').append(Math.round(task3ErrorRate * 100) / 100.0);

		System.out.println(sb.toString());		
	}

	@Override
	public void start() {

		thread = new Thread(new Runnable() {

			@Override
			public void run() {
				startTime = System.currentTimeMillis();
				while (!quit) {

					print();

					try {
						Thread.sleep(1000);
					} catch (InterruptedException e) {
						e.printStackTrace();
					}

				}

			}

		});

		thread.start();

	}

	@Override
	public void stop() {
		System.out.println("stop monitor...");
		quit = true;
		if (thread != null) {
			try {
				thread.join();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		System.out.println("stop monitor...ok");
	}

}
